package tul.cot.job;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.math.stat.descriptive.DescriptiveStatistics;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import tul.cot.MigStats;
import tul.cot.MigrationTree;
import tul.cot.MigStats.Change;
import tul.cot.exception.ParseTreeException;
import tul.cot.job.MyCounters.USER_COUNTERS;
import tul.cot.util.StatisticsUtils;
import tul.cot.util.TreeStats;
import tul.cot.util.TreeUtils;


public class InitStats extends Configured implements Tool {
	
	static enum OUTPUT_TYPE {
		START_TIME,
		SEGMENTS_NUM,
		TREE_DEPTH,
		PAIRS
	}
	
	private static final Logger LOG = Logger.getLogger(InitStats.class);
	static {
//		LOG.setLevel(Level.DEBUG);
	}
	
	public static final String PAIRS_SEP = ";";
//	static final String PAIR_SEP = "->";
	
	private static final String CONF_NUM_TREES = "migrations.min.trees.num";
	
	public static class InitMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
		
		private static final String CONF_MIN_BEFORE_START = "migrations.min.splits.before.start";
		private static final String CONF_MIN_IN_FIRST_SEG = "migrations.min.splits.first.segment";
		
		private static Counter BAD_TREES_CNT;
		
		private static int MIN_BEFORE_START = -1;
		private static int MIN_FIRST_SEGMENT = -1;
		
		private IntWritable outKey = new IntWritable();
		private Text outValue = new Text();
		
		protected void setup(Context context) throws java.io.IOException ,InterruptedException {
			Configuration conf = context.getConfiguration();
			MIN_BEFORE_START = conf.getInt(CONF_MIN_BEFORE_START, -1);
			MIN_FIRST_SEGMENT = conf.getInt(CONF_MIN_IN_FIRST_SEG, -1);
			
			if (MIN_BEFORE_START < 0) {
				throw new IllegalArgumentException("MIN_BEFORE_START < 0");
			}
			
			if (MIN_FIRST_SEGMENT <= 0) {
				throw new IllegalArgumentException("MIN_FIRST_SEGMENT <= 0");
			}
			
			BAD_TREES_CNT = context.getCounter(USER_COUNTERS.BAD_TREES);
		}
		
		protected void map(LongWritable key, Text value, Context context) throws java.io.IOException ,InterruptedException {
			
			MigrationTree tree = null;
			
			try {
				tree = new MigrationTree(value.toString());
			} catch (ParseTreeException e) {
				BAD_TREES_CNT.increment(1);
				return;
			}
			
			emit_TreeDepth(context, tree);
			emit_MigrationPairs(context, tree);
			emit_StartTime_SegmentsNum(context, tree);
		}

		private void emit_MigrationPairs(Context context, MigrationTree tree) throws IOException, InterruptedException {
			TreeStats treeStats = new TreeStats(tree, 0, 0);
			MigStats stats = treeStats.iterator().next();
			
			for (Entry<Change, Integer> e: stats.getCounters()) {
				Change change = e.getKey();
				double freq = ((double)e.getValue()) / stats.getTotalDispersals();
				
				emit_Pair(context, change, freq);
			}
		}

		private void emit_Pair(Context context, Change change, double freq) throws IOException, InterruptedException {
			String pairStr = String.format("%1$s,%2$s;%3$s", change.from, change.to, freq);
			
			outKey.set(OUTPUT_TYPE.PAIRS.ordinal());
			outValue.set(pairStr);
			
			context.write(outKey, outValue);
		}

		private void emit_StartTime_SegmentsNum(Context context, MigrationTree tree) throws IOException, InterruptedException {
			
			double startTime = -1;
			double firstSegLength = -1;
			int segNum = 0;
			
			List<MigrationTree> nodes = TreeUtils.sortNodes(tree);
			
			int curr = 0;
			MigrationTree node = curr < nodes.size() ? nodes.get(curr) : null;
			int num = 0;
			
			while (node != null && num < MIN_BEFORE_START) {
				
				num += node.getNumSons();
				node = ++curr < nodes.size() ? nodes.get(curr) : null;
			}
			
			if (num < MIN_BEFORE_START) {
				throw new IllegalArgumentException("number of migrations to small (MIN_BEFORE_START)");
			}
			
			LOG.debug(String.format("num (%d) >= MIN_BEFORE_START (%d); curr (%d)", 
					num, MIN_BEFORE_START, curr));
			LOG.debug(node);
			
			
			if (MIN_BEFORE_START == 0) {
				startTime = nodes.get(0).getTime();
			} else {
				// find next with different depth
				LOG.debug(">>> find next with different depth");
				
				double d1 = nodes.get(curr - 1).getTime();
				
//				while (node != null && MigrationTree.depthEquals(node.getTime(), d1)) {
				while (node != null && Math.abs(node.getTime() - d1) < 0.000000001) {
					node = ++curr < nodes.size() ? nodes.get(curr) : null;
				}
				
				if (node == null) {
					startTime = d1;
				} else {
					LOG.debug(String.format("found: curr (%d), depth (%f)", curr, node.getTime()));
					LOG.debug(String.format("startTime = (%f + %f) / 2;", d1, node.getTime()));
					
					startTime = (d1 + node.getTime()) / 2;
				}
				
				LOG.debug("<<< find next with different depth");
			}
			
			// emit start time
			outKey.set(OUTPUT_TYPE.START_TIME.ordinal());
			outValue.set(Double.toString(startTime));
			LOG.debug(String.format("context.write(%d, %s);", outKey.get(), outValue.toString()));
			context.write(outKey, outValue);
			
			num = 0;
			while (node != null && num < MIN_FIRST_SEGMENT) {

				num += node.getNumSons();
				node = ++curr < nodes.size() ? nodes.get(curr) : null;
			}
			
			if (num < MIN_FIRST_SEGMENT) {
				throw new IllegalArgumentException("number of migrations to small (MIN_FIRST_SEGMENT)");
			}
			
			// find next with different depth
			double d1 = nodes.get(curr - 1).getTime();
			
//			while (node != null && MigrationTree.depthEquals(node.getTime(), d1)) {
			while (node != null && Math.abs(node.getTime() - d1) < 0.000000001) {
				node = ++curr < nodes.size() ? nodes.get(curr) : null;
			}
			
			double firstSegEnd = -1;
			
			if (node == null) {
				firstSegEnd = d1;
			} else {
				firstSegEnd = (d1 + node.getTime()) / 2;
			}
			
			// NOTICE: startTime > firstSegEnd
			firstSegLength = startTime - firstSegEnd;
			
			if (firstSegLength <= 0)
				throw new RuntimeException("firstSegLength <= 0");
			
			LOG.debug(String.format("segNum = (int) (%f / %f);", startTime, firstSegLength));
			segNum = (int) (startTime / firstSegLength);
			
			// emit num of segments
			outKey.set(OUTPUT_TYPE.SEGMENTS_NUM.ordinal());
			outValue.set(Integer.toString(segNum));
			LOG.debug(String.format("context.write(%d, %s);", outKey.get(), outValue.toString()));
			context.write(outKey, outValue);
		}

		private void emit_TreeDepth(Context context, MigrationTree tree) throws IOException, InterruptedException {
			outKey.set(OUTPUT_TYPE.TREE_DEPTH.ordinal());
			outValue.set(Double.toString(tree.getTime()));
			context.write(outKey, outValue);
		}
	}
	
	public static class InitReducer extends Reducer<IntWritable, Text, Text, Text> {
		
		private static int NUM_TREES = -1;
		private static Counter BAD_PAIRS_CNT;
		
		private static Text outKey = new Text();
		private static Text outValue = new Text();
		
		private Map<Change, List<Double>> map = new HashMap<MigStats.Change, List<Double>>(300);
		
		
		protected void setup(Context context) {
			
			BAD_PAIRS_CNT = context.getCounter(USER_COUNTERS.BAD_PAIRS);
			
			Configuration conf = context.getConfiguration();
			NUM_TREES = conf.getInt(CONF_NUM_TREES, 0);
			
			if (NUM_TREES <= 0) {
				throw new IllegalArgumentException("NUM_TREES <= 0");
			}
		}
		
		protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException ,InterruptedException {

//			StringBuilder allPairs = new StringBuilder();
//			StringBuilder goodPairs = new StringBuilder();
			
			List<Change> allPairs = new ArrayList<Change>();
			List<Change> goodPairs = new ArrayList<Change>();
			
			
			OUTPUT_TYPE outputType = OUTPUT_TYPE.values()[key.get()];
			switch (outputType) {
				case TREE_DEPTH:
				case START_TIME:
				case SEGMENTS_NUM:
					DescriptiveStatistics stats = new DescriptiveStatistics();
					for (Text vt: values) {
						double value = Double.parseDouble(vt.toString());
						stats.addValue(value);
					}
					
					String statsDesc = getStatsDesc(stats);
					
					outKey.set(outputType.toString());
					outValue.set(statsDesc);
					context.write(outKey, outValue);
					
				break;
				case PAIRS:
					
					for (Text vt: values) {
						String pairStr = vt.toString();
						String[] split = pairStr.split(";");
						String changeStr = split[0];
						double freq = Double.parseDouble(split[1]);
						
						String[] changeSplit = changeStr.split(",");
						String from = changeSplit[0];
						String to	= changeSplit[1];
						Change change = new Change(from, to);
						
						addFreqValue(change, freq);
					}
					
					for (Entry<Change, List<Double>> e: map.entrySet()) {
						Change change = e.getKey();
						
						// all pairs
						allPairs.add(change);
						
						// reasonable pairs (non-zero median)
						List<Double> freqList = e.getValue();
						Collections.sort(freqList);
						
						int leadingZeros = NUM_TREES - freqList.size();
						double freqMedian = StatisticsUtils.getMedian(freqList, leadingZeros, 0);
						
						if (freqMedian > 0) {
							goodPairs.add(change);
						} else {
							LOG.debug(String.format("Bad pair: %1$s\tnon-zero-freqs: %2$d", change, freqList.size()));
							BAD_PAIRS_CNT.increment(1);
						}
					}
					
					// emit all pairs
					outKey.set(outputType.toString() + "_ALL");
					outValue.set(getPairsString(allPairs));
					context.write(outKey, outValue);
					
					// emit good pairs
					outKey.set(outputType.toString() + "_GOOD");
					outValue.set(getPairsString(goodPairs));
					context.write(outKey, outValue);
					
				break;
				default:
					throw new RuntimeException("Unknown output type");
			}
		}

		private String getPairsString(List<Change> pairsList) {
			StringBuilder sb = new StringBuilder();
			for (Change change: pairsList) {
				addPair(sb, change);
			}
			
			return sb.toString();
		}

		private void addPair(StringBuilder pairs, Change change) {
			if (pairs.length() > 0) {
				pairs.append(PAIRS_SEP);
			}
			pairs.append(change.toString());
		}

		private void addFreqValue(Change change, double freq) {
			List<Double> freqList = map.get(change);
			if (freqList == null) {
				freqList = new ArrayList<Double>();
				freqList.add(freq);
				map.put(change, freqList);
			} else {
				freqList.add(freq);
			}
		}

		private String getStatsDesc(DescriptiveStatistics stats) {
			double min = stats.getMin();
			double max = stats.getMax();
			double avg = stats.getMean();
			double median = stats.getPercentile(50);
			
			StringBuilder sb = new StringBuilder();
			sb.append("MIN=");
			sb.append(min);
			sb.append(";");
			sb.append("MAX=");
			sb.append(max);
			sb.append(";");
			sb.append("AVG=");
			sb.append(avg);
			sb.append(";");
			sb.append("MEDIAN=");
			sb.append(median);
			
			return sb.toString();
		}
	}
	
	@Override
	public int run(String[] args) throws Exception {
		
		if (args.length != 5) {
			System.err.println("usage: <numTrees> <minBeforeStart> <minFirstSegment> <in> <out>");
			return -1;
		}
		
		int numTrees = Integer.parseInt(args[0]);
		int minBeforeStart = Integer.parseInt(args[1]);
		int minFirstSegment = Integer.parseInt(args[2]);
		Path inPath = new Path(args[3]);
		Path outPath = new Path(args[4]);
		
		if (minBeforeStart < 0) {
			throw new IllegalArgumentException("minBeforeStart < 0");
		}
		
		if (minFirstSegment <= 0) {
			throw new IllegalArgumentException("minFirstSegment <= 0");
		}
		
		getConf().setInt(CONF_NUM_TREES, numTrees);
		getConf().setInt(InitMapper.CONF_MIN_BEFORE_START, minBeforeStart);
		getConf().setInt(InitMapper.CONF_MIN_IN_FIRST_SEG, minFirstSegment);
		
		String jobName = String.format("%s: numTrees(%d) minBeforeStart(%d) minFirstSegment(%d) in(%s) out(%s)",
				getClass().getSimpleName(), numTrees, minBeforeStart, minFirstSegment, inPath.getName(), outPath.getName());
		Job job = new Job(getConf(), jobName);
		job.setJarByClass(getClass());
		
		FileInputFormat.setInputPaths(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		
		job.setMapperClass(InitMapper.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setReducerClass(InitReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		int ret = ToolRunner.run(new InitStats(), args);
		System.exit(ret);
	}

}
